package com.xxx.redismq.product.controller;

import java.text.SimpleDateFormat;
import java.util.UUID;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.xxx.redismq.common.pojo.Message;
import com.xxx.redismq.common.utils.RedisMqUtil;

/**
 * 利用 redis 实现延迟消息队列——消息发送方
 * 实现原理：当一条Message消息发送过来时，以String类型存储到redis数据库中。
 * 		在没有被消费之前，就将key放到zset类型中排队，称之为消息队列。
 * 		然后将排好队的消息取出来，按顺序放进List当中，消费者实际上消费的消息都是从list中取出来的。
 */
@RestController
public class SendMsgController {

    @Resource
    private RedisMqUtil redisMQ;
    
    @Value("${mq.queue.first}")
    private String MQ_QUEUE_FIRST;

    /**
     * 发送单条消息
     * http://localhost:8082/postMsg?msg=发送消息1
     */
    @RequestMapping("/postMsg")
    public Message postMsg(@RequestParam(value = "msg", defaultValue = "测试默认消息") String msg) {
        Message message = new Message();
        message.setTopic("msg");
        message.setId(UUID.randomUUID().toString());
        message.setDelay(20000); // 延时20s（单位是ms）
        message.setPriority(0);
        message.setTtl(20 * 60); // 设置消息池ttl，防止长期占用内存，这里设置为20分钟
        message.setBody(msg);
        message.setCreateTime(System.currentTimeMillis());
        message.setStatus(0); // 状态为延时中

        // 将消息存入redis的消息池中
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String now = sdf.format(message.getCreateTime());
        Long consumedTime = message.getCreateTime() + message.getDelay();
        System.out.println("发送了消息："+message.getBody()+"，当前时间：" + now +", 延时时间："+message.getDelay()/1000+"s， 消费应被消费的时间：" + sdf.format(consumedTime));
        redisMQ.addMsgIntoPool(message);
        
        // 将消息放到RedisMQ中排队（即向zset中添加消息，key为队列名称（queue:1），score为createTime+delay+priority，value为消息ID）
        redisMQ.addMsgIntoMq(MQ_QUEUE_FIRST, // zset的key
        		message.getCreateTime() + message.getDelay() + message.getPriority(), // zset的score
        		message.getId()); 	// zset的member为消息ID

       return message;
    }
    
    /**
     * 批量发送消息
     * http://localhost:8082/postMsgs
     */
    @RequestMapping("/postMsgs")
    public String postMsgs() throws InterruptedException {
    	for(int i=1; i<=5; i++) {
    		String msg = "测试默认消息" + i;
    		Thread.sleep(2000);
    		postMsg(msg);
    	}
    	
       return "success";
    }

}
